package com.yunda.nebula.gateway.sync.http;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yunda.nebula.gateway.core.DataChangedListener;
import com.yunda.nebula.gateway.core.SyncDataService;
import com.yunda.nebula.gateway.core.entity.ConfigData;
import com.yunda.nebula.gateway.core.enums.DataEventTypeEnum;
import com.yunda.nebula.gateway.core.exception.GateWayException;
import com.yunda.nebula.gateway.core.utils.NebulaThreadFactory;
import lombok.Data;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.client.OkHttp3ClientHttpRequestFactory;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * \* User: wei.zhang
 * \* Date: 2020/9/16 10:19
 * \* Description:
 *
 * @author wei.zhang
 * \
 */
@Slf4j
@Data
public class HttpSyncDataService implements SyncDataService {

    private static final AtomicBoolean RUNNING = new AtomicBoolean(false);
    private HttpConfig httpConfig;
    private ExecutorService executorService;

    private String server;
    /**
     * only use for http long polling.
     */
    private RestTemplate httpClient;

    private DataChangedListener dataChangedListener;

    private Map<String,HttpTask> taskMap = new ConcurrentHashMap<>(16);

    /**
     * default: 10s.
     */
    private Duration connectionTimeout = Duration.ofSeconds(10);
    private DataRefreshFactory factory;

    private List<String> appIds = new CopyOnWriteArrayList<>();

    public HttpSyncDataService(HttpConfig httpConfig,DataChangedListener dataChangedListener) {
        this.httpConfig = httpConfig;
        this.executorService = Executors.newCachedThreadPool(NebulaThreadFactory.create("http-polling",true));
        this.dataChangedListener = dataChangedListener;
        this.server = httpConfig.getUrl();
        factory = new DataRefreshFactory();
        this.start(httpConfig);
    }

    public void start(HttpConfig httpConfig) {

        OkHttp3ClientHttpRequestFactory factory = new OkHttp3ClientHttpRequestFactory();
        factory.setConnectTimeout((int) this.connectionTimeout.toMillis());
        factory.setReadTimeout((int) HttpConstants.CLIENT_POLLING_READ_TIMEOUT);
        this.httpClient = new RestTemplate(factory);
        // It could be initialized multiple times, so you need to control that.
        if (RUNNING.compareAndSet(false, true)) {
            // fetch all group configs.
            List<String> apps = this.fetchAllApps();
            appIds.addAll(apps);
            this.fetchAllConfig();
            AppRefreshTask appRefreshTask = new AppRefreshTask();
            appRefreshTask.setSyncDataService(this);
            for (String appIds:apps){
                this.createTask(appIds);
            }
            this.executorService.execute(appRefreshTask);
        } else {
            log.info("nebula gateway http long polling was started, executor=[{}]", executorService);
        }
    }

    public void fetchAllConfig() {
        for (int index = 0; index < this.appIds.size(); index++) {
            String app = appIds.get(index);
            try {
                this.doFetchAppConfig(app, DataEventTypeEnum.CREATE);
            } catch (GateWayException e) {
                log.error("获取app信息信息报错 !!!",e);
            }
        }
    }

    @SneakyThrows
    @Override
    public void doFetchAppConfig(String app, DataEventTypeEnum eventTypeEnum) {
        StringBuilder params = new StringBuilder();
        String url = server + "/nebula/gateway/config/fetch.do?appId=" + app;
        log.info("request configs: [{}]", url);
        String json = null;
        try {
            json = this.httpClient.getForObject(url, String.class);
        } catch (RestClientException e) {
            String message = String.format("fetch config fail from server[%s], %s", url, e.getMessage());
            log.warn(message);
            throw new GateWayException(message, e);
        }

        // update local cache
        boolean updated = this.updateCacheWithJson(json);
        if (!updated) {
            log.info("The config of the server[{}] has not been updated or is out of date. Wait for 30s to listen for changes again.", server);
            TimeUnit.SECONDS.sleep(10);
            return;
        }
        log.info("get latest configs: [{}]", json);
        JSONObject jsonObject = JSONObject.parseObject(json);
        ConfigData configData = JSONObject.parseObject(jsonObject.getString("data"),ConfigData.class);
        dataChangedListener.onAppChanged(configData,eventTypeEnum);
        // not updated. it is likely that the current config server has not been updated yet. wait a moment.
    }

    @Override
    public void doRemoveAppConfig(String appId, DataEventTypeEnum typeEnum){
        ConfigData configData = new ConfigData();
        configData.setAppId(appId);
        log.info("删除任务，任务ID 为 {}",appId);
        boolean ret = factory.delete(JSONObject.parseObject(JSONObject.toJSONString(configData)));
        removeTask(appId);
        log.info("开始删除任务 清除缓存");
        log.info("删除任务，任务ID 为 {}",appId);
        dataChangedListener.onAppChanged(configData,typeEnum);

    }


    public boolean updateCacheWithJson(String json) {
        JSONObject jsonObject = JSONObject.parseObject(json);
        JSONObject data = jsonObject.getJSONObject("data");
        return factory.update(data);
    }

    @Override
    public List<String> fetchAllApps() {
        //http://127.0.0.1:8800
        String url = server + "/nebula/gateway/config/app.do";
        log.info("request configs: [{}]", url);
        String json = null;
        try {
            json = this.httpClient.getForObject(url, String.class);
        } catch (RestClientException e) {
            String message = String.format("fetch config fail from server[%s], %s", url, e.getMessage());
            log.warn(message);
            throw new GateWayException(message, e);
        }
        String result = JSONObject.parseObject(json).getString("data");
        List<String> appIds = JSONArray.parseArray(result,String.class);
        return appIds;
    }

    public void removeTask(String app) {
        HttpTask t = taskMap.get(app);
        t.stop();
    }


    public void createTask(String app) {
        HttpLongPollingTask task = new HttpLongPollingTask(server,this);
        task.setAppId(app);
        task.setRestTemplate(httpClient);
        task.setFactory(factory);
        task.setUrl(server);
        taskMap.put(app,task);
        executorService.execute(task);
    }


}
